package com.tdy.cdc.function;

import com.alibaba.fastjson.JSONObject;
import com.tdy.cdc.util.JDBCUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @author NanHuang
 * @Date 2023/3/21
 */
public class MySQLSinkFunction extends RichSinkFunction<JSONObject> {

    private Connection mysqlConn;
    private PreparedStatement ps;

    @Override
    public void open(Configuration parameters) throws Exception {
        mysqlConn = JDBCUtil.getCdrConnection();
    }

    @Override
    public void close() throws Exception {
        if (mysqlConn != null) {
            mysqlConn.close();
        }
        if (ps != null) {
            ps.close();
        }
    }

    @Override
    public void invoke(JSONObject obj, Context context) throws Exception {
        /*String op = obj.getString("op");
        // 1 拼接sql
        switch (op){
            case "r":
            case "c":
            case "u":
            case "d":
                // TODO 根据不同操作类型对数据做处理
                insertData(obj);
                break;
        }*/
        insertData(obj);
    }

    private void deleteData(JSONObject obj) {
        String template = "delete from  %s  where unique_id = ?";
        String sql = String.format(template, obj.getString("table_name").substring(4));
        try {
            PreparedStatement ps = mysqlConn.prepareStatement(sql);
            ps.setString(1,obj.getString("unique_id"));
            ps.execute();
        } catch (SQLException e) {
            throw new RuntimeException(e.getMessage());
        }
    }

    private void insertData(JSONObject obj) {
        String template = "replace into %s (%s) values (%s)";
        String targetTable = obj.getString("table_name").substring(4);
        Set<String> keys = obj.keySet();
        keys.removeIf(key -> "".equals(obj.getString(key)) || "table_name".equals(key) || "op".equals(key));
        ArrayList<String> columns = new ArrayList<>(obj.keySet());
        String columnStr = String.join(",", columns);
        String sql = String.format(template, targetTable, columnStr, columnStr.replaceAll("[^,]+", "?"));
        try {
            // 2 预编译
            ps = mysqlConn.prepareStatement(sql);
            // 3 填充占位符
            for (int i = 0; i < columns.size(); i++) {
                String value = obj.getString(columns.get(i));
                if ("false".equals(value) || "FALSE".equals(value)){
                    value = "0";
                }
                if ("true".equals(value) || "TRUE".equals(value)){
                    value = "1";
                }
                ps.setString(i + 1,value);
            }
            // 4 运行sql
            ps.execute();
        } catch (SQLException e) {
            throw new RuntimeException(e.getMessage());
        }
    }
}
